package com.scala.CopeWithData

import java.util.UUID

import com.java.Enum.{CurrentState, StoreBuyWay, StoreType}
import com.java.config.IpConfig
import com.java.util.{InitialUtil, PatternUtil}
import com.scala.CaseClass.RentalBuildingInfo
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode, SparkSession}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

/**
 * @Author: ZhangJin
 * @Date: 2020/9/20 21:57
 */
// TODO:   数据加载进入hive后,进行处理
@Service("CopeWithDataForRentInfo")
class CopeWithDataForRentInfo @Autowired()(val session: SparkSession,val ipConfig:IpConfig) extends Serializable{


  def copeWithDataForRent(fileName: String,SQLforTable: String): Unit = {
    import session.implicits._
    val hc: SQLContext = session.sqlContext
    // TODO:   去除null、NaN
    val DF: DataFrame = hc.sql("select * from default."+fileName).as[RentalBuildingInfo].rdd.
      map(each => {
        /*   设置数据处理指标  */
        val UUid =UUID.randomUUID().toString
        val location = each.Local
        val LocationOfLatAndLong = InitialUtil.getcoorder(if(each.LocationDetail.contains(each.Local)) each.LocationDetail else each.Local + each.LocationDetail)
        val AreaOfHouse = each.AreaOfHouse
        val LocationDetail= each.LocationDetail
        val Depth = each.Depth
        val Width = each.Width
        val Sheer = each.Sheer
        val Type = StoreType.getType(each.Type).getType
        val Introduction = each.Introduction
        val StateNow = CurrentState.getState(each.StateNow).getType
        val WebSite = each.WebSite
        val Title = StoreBuyWay.getWay(each.Title).getType
        /* 受欢迎程度 */
        val NumOfSeePepple=PatternUtil.PopularityAnalysis(each.NumOfSeePepple)
        val PriceByMonth=PatternUtil.PriceTransfer(PatternUtil.deleteString(each.PriceByMonth))
        RentalBuildingInfo(UUid,location,Title,NumOfSeePepple,PriceByMonth,AreaOfHouse,Width,Depth,Sheer,Type,StateNow,LocationDetail,Introduction,WebSite,LocationOfLatAndLong)
      }).toDF().na.drop().where("AreaOfHouse  <> '' ").where("Width  <> '' ").where("Depth  <> '' ").where("PriceByMonth  <> '' ")
    // TODO:   处理后的数据进行持久化
    hc.sql("CREATE DATABASE IF NOT EXISTS CopeWithData")
    hc.sql("use CopeWithData")
    DF.write.mode(SaveMode.Overwrite).parquet("hdfs://"+ipConfig.getIp+":"+ipConfig.getPort+"/usr/data/codeLaterRentInfoData"+ fileName +".parquet")
    hc.sql("create table if not exists codeLaterRentInfoData"+ fileName + "(UUid String,Local String,Title String,NumOfSeePepple String,PriceByMonth String,AreaOfHouse String,Width String,Depth String,Sheer String,Type String,StateNow String,LocationDetail String,Introduction String,WebSite String,LocationOfLatAndLong String) stored as parquet")
    hc.sql("load data inpath 'hdfs://" + ipConfig.getIp+":"+ipConfig.getPort+ "/usr/data/codeLaterRentInfoData"+ fileName +".parquet' overwrite into table codeLaterRentInfoData" + fileName)

  }

}


